package com.shensi.appCollect.task;

import com.alibaba.fastjson.JSONObject;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.conditions.update.LambdaUpdateWrapper;
import com.shensi.appCollect.constantEnum.BatchTaskCons;
import com.shensi.appCollect.enums.BatchTaskStatus;
import com.shensi.appCollect.pojo.AppUserInfo;
import com.shensi.appCollect.pojo.BatchSubtask;
import com.shensi.appCollect.pojo.BatchTask;
import com.shensi.appCollect.pojo.dto.SendWxTemplateMessageRunnableParam;
import com.shensi.appCollect.pojo.wx.WxTemplateMessageSendRequest;
import com.shensi.appCollect.service.AppUserInfoService;
import com.shensi.appCollect.service.BatchSubtaskService;
import com.shensi.appCollect.service.BatchTaskService;
import com.shensi.appCollect.utils.JsonUtils;
import com.shensi.appCollect.utils.MyLambadaUpdateWrapper;
import com.shensi.appCollect.utils.weixinapp.WxPushUtil;
import com.xxl.job.core.handler.annotation.XxlJob;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.annotation.Bean;
import org.springframework.scheduling.concurrent.CustomizableThreadFactory;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.time.LocalDateTime;
import java.util.List;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.atomic.AtomicBoolean;

/**
 * @date 2023/11/20
 */
@Component
public class BatchTaskExecutor {

    private static final Logger log = LoggerFactory.getLogger(BatchTaskExecutor.class);

    private static final int WORK_QUEUE_SIZE = 1024;

    @Autowired
    private BatchTaskService batchTaskService;
    @Autowired
    private BatchSubtaskService batchSubtaskService;
    @Autowired
    private AppUserInfoService appUserInfoService;
    @Autowired
    private WxPushUtil wxPushUtil;
    private final AtomicBoolean queueLock = new AtomicBoolean(false);

    private final BlockingQueue<Runnable> workQueue = new LinkedBlockingQueue<>(WORK_QUEUE_SIZE);

    @Autowired
    @Qualifier("batchTaskThreadPoolExecutor")
    private ThreadPoolTaskExecutor threadPoolExecutor;

    @Bean
    public ThreadPoolTaskExecutor batchTaskThreadPoolExecutor() {
        MyThreadPoolTaskExecutor taskExecutor = new MyThreadPoolTaskExecutor();
        taskExecutor.setCorePoolSize(8);
        taskExecutor.setMaxPoolSize(8);
        taskExecutor.setKeepAliveSeconds(90);
        taskExecutor.setBlockingQueue(workQueue);
        taskExecutor.setThreadFactory(new CustomizableThreadFactory("batch-subtask-"));
        taskExecutor.setRejectedExecutionHandler(new ThreadPoolExecutor.AbortPolicy());
        return taskExecutor;
    }

    /**
     * 每分钟尝试获取最早的任务并执行
     * 多个任务，按顺序执行（每隔1分钟）0 * * * * ?
     */
    @XxlJob("scheduledRunTask")
    public void scheduledRunTask() {
        log.info("[scheduledRunTask]定时任务开始执行！");
        runEarlySubtask();
    }


    public void runEarlySubtask() {
        // 每次获取多个任务，按顺序启动
        List<BatchSubtask> earlyTasks = getEarlySubtasks(10);
        for (BatchSubtask earlyTask : earlyTasks) {
            // 当返回==1才继续尝试循环启动，其它情况跳出
            if (runSubtask(earlyTask) != 1) break;
        }
    }

    private List<BatchSubtask> getEarlySubtasks(int size) {
        // 先查看有没有意外终止没有执行完的，状态是进行中，且开始时间是3个小时之前的
        LambdaQueryWrapper<BatchSubtask> wrapper = new LambdaQueryWrapper<>();
        wrapper.isNotNull(BatchSubtask::getStartTime)
                .lt(BatchSubtask::getStartTime, LocalDateTime.now().minusHours(3))
                .eq(BatchSubtask::getSubtaskStatus, BatchTaskStatus.RUNNING)
                .orderByAsc(BatchSubtask::getId)
                .last("limit " + size);
        List<BatchSubtask> incompleteTasks = batchSubtaskService.list(wrapper);
        int incompleteSize = incompleteTasks.size();
        if (incompleteSize == size) {
            return incompleteTasks;
        }
        // 查找最就开始的任务
        wrapper = new LambdaQueryWrapper<>();
        wrapper.isNull(BatchSubtask::getStartTime)
                .eq(BatchSubtask::getSubtaskStatus, BatchTaskStatus.NOT_STARTED)
                .orderByAsc(BatchSubtask::getId)
                .last("limit " + (size - incompleteSize));
        List<BatchSubtask> notStartTasks = batchSubtaskService.list(wrapper);
        if (notStartTasks.isEmpty()) {
            return incompleteTasks;
        }
        incompleteTasks.addAll(notStartTasks);
        return incompleteTasks;
    }

    /**
     * @param subtask
     * @return 1-循环执行下一个任务，其它跳出循环
     */
    private int runSubtask(BatchSubtask subtask) {
        if (threadPoolExecutor.getThreadPoolExecutor().isShutdown()) {
            return 2;
        }
        //log.info("尝试启动子任务,{}",subtask);
        if (!checkQueueSizeEnough(subtask.getItemNum())) {
            return 2;
        }
        if (!queueLock.compareAndSet(false, true)) {
            return 2;
        }
        int r = addSubtaskToQueue(subtask);
        queueLock.set(false);
        return r;
    }

    private int addSubtaskToQueue(BatchSubtask subtask) {
        //log.info("开始启动子任务,{}",subtask.getId());
        MyLambadaUpdateWrapper<BatchSubtask> updateWrapper = new MyLambadaUpdateWrapper<>();
        updateWrapper.incrementSet(BatchSubtask::getVersion, 1)
                .set(BatchSubtask::getSubtaskStatus, BatchTaskStatus.RUNNING)
                .set(BatchSubtask::getStartTime, LocalDateTime.now())
                .eq(BatchSubtask::getVersion, subtask.getVersion())
                .eq(BatchSubtask::getId, subtask.getId());
        if (!batchSubtaskService.update(updateWrapper)) {
            // 更新任务状态失败，返回
            log.info("子任务启动失败,乐观锁获取失败,subtaskId:[{}]", subtask.getId());
            return 1;
        }
        subtask.setVersion(subtask.getVersion() + 1);
        BatchTask batchTask = batchTaskService.getById(subtask.getBatchId());
        if (batchTask == null) {
            updateWrapper.incrementSet(BatchSubtask::getVersion, 1)
                    .set(BatchSubtask::getSubtaskStatus, BatchTaskStatus.FAIL)
                    .set(BatchSubtask::getCompleteTime, LocalDateTime.now())
                    .eq(BatchSubtask::getVersion, subtask.getVersion())
                    .eq(BatchSubtask::getId, subtask.getId());
            if (batchSubtaskService.update(updateWrapper)) {
                subtask.setVersion(subtask.getVersion() + 1);
            }
            log.info("子任务启动失败,batchTask为null,id:[{}]", subtask.getId());
            return 1;
        }
        log.info("子任务启动成功,id:[{}]", subtask.getId());
        int r = 0;
        if (StringUtils.equals(BatchTaskCons.TASK_TYPE_SEND_WX_TEMPLATE_MESSAGE, subtask.getTaskType())) {
            r = sendWxTemplateMessageHandler(subtask, batchTask);
        }
        return r;
    }

    private int sendWxTemplateMessageHandler(BatchSubtask subtask, BatchTask batchTask) {
        LambdaUpdateWrapper<BatchSubtask> updateWrapper = new LambdaUpdateWrapper<>();
        updateWrapper.set(BatchSubtask::getSuccessNum, 0)
                .set(BatchSubtask::getFailNum, 0)
                .set(BatchSubtask::getSkipNum, 0)
                .eq(BatchSubtask::getId, subtask.getId());
        batchSubtaskService.update(updateWrapper);
        subtask.setSuccessNum(0);
        subtask.setFailNum(0);
        subtask.setSkipNum(0);
        String subtaskParam = subtask.getSubtaskParam();
        JSONObject param = JSONObject.parseObject(subtaskParam);
        Integer offset = param.getInteger("offset");
        Integer limit = param.getInteger("limit");
        if (offset == null || limit == null) {
            BatchSubtask update = new BatchSubtask();
            update.setId(subtask.getId());
            update.setSubtaskStatus(BatchTaskStatus.FAIL);
            update.setCompleteTime(LocalDateTime.now());
            batchSubtaskService.updateById(update);
            return 1;
        }
        String taskContext = batchTask.getTaskContext();
        SendWxTemplateMessageRunnableParam runnableParam = null;
        try {
            runnableParam = JsonUtils.jsonToObject(taskContext, SendWxTemplateMessageRunnableParam.class);
        } catch (IOException e) {
            log.error(e.getMessage(), e);
        }
        if (runnableParam == null || runnableParam.getSendRequest() == null) {
            return 1;
        }
        String appId = runnableParam.getGzhAppId();
        String appSecret = runnableParam.getGzhAppSecret();
        Integer enterpriseId = runnableParam.getEnterpriseId();
        WxTemplateMessageSendRequest sendRequest = runnableParam.getSendRequest();
        AppUserInfo info = new AppUserInfo();
        info.setStationId(enterpriseId);
        List<AppUserInfo> userInfos = appUserInfoService.queryPageByExample(info, offset, limit);
        int r = 0;
        if (checkQueueSizeEnough(subtask.getItemNum())) {
            int skip = subtask.getItemNum() - userInfos.size();
            int i = 0;
            for (AppUserInfo userInfo : userInfos) {
                i++;
                if (StringUtils.isNotBlank(userInfo.getBlogOpenid())) {
                    WxTemplateMessageSendRequest request = new WxTemplateMessageSendRequest();
                    request.setTouser(userInfo.getBlogOpenid());
                    request.setTemplateId(sendRequest.getTemplateId());
                    request.setUrl(sendRequest.getUrl());
                    request.setData(sendRequest.getData());
                    SendWxTemplateMessageRunnableParam param1 = new SendWxTemplateMessageRunnableParam();
                    param1.setSendRequest(request);
                    param1.setIndex(i);
                    param1.setGzhAppId(appId);
                    param1.setGzhAppSecret(appSecret);
                    SendWxTemplateMessageSubtaskRunnable runnable = new SendWxTemplateMessageSubtaskRunnable(wxPushUtil, this, batchSubtaskService, param1, subtask);
                    this.threadPoolExecutor.execute(runnable);
                } else {
                    skip++;
                }
            }
            log.info("子任务开始处理数据,待处理数据条数:[{}]", i - skip);
            if (skip > 0) {
                boolean finished;
                synchronized (subtask) {
                    subtask.setSkipNum(subtask.getSkipNum() + skip);
                    finished = subtask.getItemNum() <= (subtask.getSuccessNum() + subtask.getFailNum() + subtask.getSkipNum());
                }
                MyLambadaUpdateWrapper<BatchSubtask> wrapper = new MyLambadaUpdateWrapper<>();
                wrapper.incrementSet(BatchSubtask::getSkipNum, skip);
                if (finished) {
                    r = 1;
                    wrapper.set(BatchSubtask::getSubtaskStatus, BatchTaskStatus.SUCCESS)
                            .set(BatchSubtask::getCompleteTime, LocalDateTime.now());
                    log.info("子任务跳过条数等于总条数,执行成功,id:[{}],itemNum:[{}],skipNum:[{}]", subtask.getId(), subtask.getItemNum(), subtask.getSkipNum());
                }
                wrapper.eq(BatchSubtask::getId, subtask.getId());
                batchSubtaskService.update(wrapper);
            }
        } else {
            LambdaUpdateWrapper<BatchSubtask> wrapper = new LambdaUpdateWrapper<>();
            wrapper.set(true, BatchSubtask::getStartTime, null)
                    .set(true, BatchSubtask::getCompleteTime, null)
                    .set(BatchSubtask::getSubtaskStatus, BatchTaskStatus.NOT_STARTED)
                    .eq(BatchSubtask::getId, subtask.getId());
            batchSubtaskService.update(wrapper);
            log.info("队列空余长度不足,子任务置为初始状态,{}", subtask);
        }
        return r;
    }

    private boolean checkQueueSizeEnough(int expectAddSize) {
        return (WORK_QUEUE_SIZE - workQueue.size() - 100) > expectAddSize;
    }


    private static class MyThreadPoolTaskExecutor extends ThreadPoolTaskExecutor {
        private BlockingQueue<Runnable> blockingQueue;

        public BlockingQueue<Runnable> getBlockingQueue() {
            return blockingQueue;
        }

        public void setBlockingQueue(BlockingQueue<Runnable> blockingQueue) {
            this.blockingQueue = blockingQueue;
        }

        @Override
        protected BlockingQueue<Runnable> createQueue(int queueCapacity) {
            return this.blockingQueue;
        }
    }

}
